package com.fulu.game.bi.hive;

import com.aliyun.oss.model.AppendObjectRequest;
import com.aliyun.oss.model.AppendObjectResult;
import com.aliyun.oss.model.ObjectMetadata;
import com.fulu.game.bi.common.util.EmailService;
import com.fulu.game.bi.common.util.OssUtil;
import com.fulu.game.bi.entity.ExportBO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Repository;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;

import java.io.ByteArrayInputStream;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;

@Repository
@Slf4j
public class HivePlayeeExportDao {

    @Autowired
    private JdbcTemplate hiveJdbcTemplate;
    @Autowired
    private OssUtil ossUtil;
    @Autowired
    private EmailService emailService;

    @Async
    public void eventCondition(ExportBO bo) {
        String refreshSql = "refresh " + bo.getTableName();
        hiveJdbcTemplate.execute(refreshSql);
        //拼接sql
        StringBuilder sql = new StringBuilder();
        sql.append("SELECT * FROM " + bo.getTableName() + " where 1 = 1 ");
        if (!StringUtils.isEmpty(bo.getCondition())) {
            sql.append(bo.getCondition());
        }
        log.info("导出sql:{}", sql.toString());
        final boolean[] hasResult = {false};
        try {
            hiveJdbcTemplate.query(con -> {
                PreparedStatement preparedStatement =
                        con.prepareStatement(sql.toString(),
                                ResultSet.TYPE_FORWARD_ONLY,
                                ResultSet.CONCUR_READ_ONLY);
                preparedStatement.setFetchSize(Integer.MIN_VALUE);
                preparedStatement.setFetchDirection(ResultSet.FETCH_FORWARD);
                return preparedStatement;
            }, rs -> {
                hasResult[0] = true;
                StopWatch stopWatch = new StopWatch();
                stopWatch.start("首次上传");
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                StringBuilder columnNames = new StringBuilder();
                for (int i = 1; i <= columnCount; i++) {
                    String columnComment = bo.getColumnMap().get(metaData.getColumnName(i));
                    columnNames.append(columnComment == null ? metaData.getColumnName(i) + "," : columnComment + ",");
                }
                columnNames.delete(columnNames.length() - 1, columnNames.length());
                columnNames.append("\n");
                for (int i = 1; i <= columnCount; i++) {
                    columnNames.append("\"" + rs.getObject(i) + "\",");
                }
                columnNames.delete(columnNames.length() - 1, columnNames.length());
                columnNames.append("\n");
                ObjectMetadata meta = new ObjectMetadata();
                AppendObjectRequest appendObjectRequest = new AppendObjectRequest(ossUtil.getBucketName(), bo.getKey(),
                        new ByteArrayInputStream(columnNames.toString().getBytes()), meta);
                appendObjectRequest.setPosition(0L);
                AppendObjectResult appendObjectResult = OssUtil.ossClient.appendObject(appendObjectRequest);
                stopWatch.stop();
                log.info("上传时间{}", stopWatch.prettyPrint());
                StringBuilder resultSb = new StringBuilder();
                int count = 1000;
                while (rs.next()) {
                    count--;
                    appendRow(rs, columnCount, resultSb);
                    if (count <= 0) {
                        log.debug("row:{}", resultSb.toString());
                        appendObjectRequest.setPosition(appendObjectResult.getNextPosition());
                        appendObjectRequest.setInputStream(new ByteArrayInputStream(resultSb.toString().getBytes()));
                        appendObjectResult = OssUtil.ossClient.appendObject(appendObjectRequest);
                        log.debug("上传成功");
                        resultSb.delete(0, resultSb.length());
                        count = 1000;
                    }
                }
                if (resultSb.length() > 0) {
                    appendObjectRequest.setPosition(appendObjectResult.getNextPosition());
                    appendObjectRequest.setInputStream(new ByteArrayInputStream(resultSb.toString().getBytes()));
                    OssUtil.ossClient.appendObject(appendObjectRequest);
                }
            });
            if (hasResult[0]) {
                emailService.sendHtmlMail(bo.getEmail(), ossUtil.getHost() + bo.getKey());
            } else {
                emailService.sendSimpleMail(bo.getEmail(), bo.getFileName() + "导出，筛选条件没有内容可导出");
            }
            log.info("导出成功，filename：{}", bo.getFileName());
        } catch (Exception e) {
            log.error("导出失败，filename：{}", bo.getFileName(), e);
            emailService.sendSimpleMail(bo.getEmail(), bo.getFileName() + "导出失败，请联系邓卓或者重新导出");
        }

    }

    private void appendRow(ResultSet rs, int columnCount, StringBuilder resultSb) throws SQLException {
        for (int i = 1; i <= columnCount; i++) {
            resultSb.append("\"" + rs.getObject(i) + "\",");
        }
        resultSb.delete(resultSb.length() - 1, resultSb.length());
        resultSb.append("\n");
        log.debug("csv:{}", resultSb);
    }

}
